package com.haredot.controller;

import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.haredot.constraint.RabbitConst;
import com.haredot.entity.Notify;
import com.haredot.service.NotifyService;
import com.haredot.vo.R;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.http.MediaType;
import org.springframework.util.Assert;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import javax.annotation.Resource;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 *  text/event-stream 媒体格式：
 *
 *     基于 http 的 服务端 单工通信技术 （服务器 会主动推送消息 给客户端）
 *    多应用于  通知
 *
 *  SseEmitter 可以使用 text/event-stream 流数据的处理
 *
 *  SseEmitter : 服务端 向 客户端 推送消息 的 通道
 *
 *  工作原理：
 *
 *      a)  客户端 订阅 SseEmitter 推送消息的  事件流
 *           要求返回的类型必须是 text/event-stream
 *
 *      b)  服务端 推送 消息
 *
 */
@RestController
@RequestMapping("/notify")
@Slf4j
public class NotifyController {

    private final Map<Long, SseEmitter> sseEmitterMap = new ConcurrentHashMap<>();

    @Resource
    private NotifyService notifyService ;

    /**
     * 接收通知队列中的消息，并将消息投递到 SseEmitter 中
     */
    @RabbitListener(queues = RabbitConst.NOTIFY_QUEUE)
    public void notifyPublisher(Notify notify) throws Exception{
        // 获取通知的 用户ID
        Long userId = notify.getUserId();
        // 根据用户ID , 获取 对应的 SseEmitter
        SseEmitter sseEmitter = sseEmitterMap.get(userId);
        // 找到，说明用户在线，否则说明用户离线
        if (sseEmitter != null) {
            // 发送通知
            sseEmitter.send(SseEmitter.event().data(notify));
        }
    }

    /**
     * 订阅消息流
     * @param userId
     * @return
     */
    @GetMapping(value = "/subscribe/{uid}", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public SseEmitter subscribe(@PathVariable("uid") Long userId) {
        Assert.notNull(userId, "userID 值为空");
        // SseEmitter 服务单向通道，默认链接的超时间 是 30秒， 可以 设置超时时间为0， 代表不限制超时时间。
        SseEmitter sseEmitter = new SseEmitter(0L);
        // 绑定和用户的关系
        sseEmitterMap.put(userId, sseEmitter) ;
        // 当 通信完成、可以将 单向通道 从 MAP 中移除
        sseEmitter.onCompletion(()-> sseEmitterMap.remove(userId));
        // 当 通信产生异常 、怎么办 ？
        sseEmitter.onError(e -> log.error(e.getMessage()));
        return sseEmitter ;
    }

    /**
     * 查询未读信息数量
     * @return
     */
    @GetMapping("/unReadMessage")
    public R<Map<String, Long>> queryUnReadMessage() {
        return notifyService.queryUnReadMessages();
    }

    /**
     * 查询通知信息
     */
    @GetMapping("/list")
    public R<Page<Notify>> queryNotifys(@RequestParam(value = "page", required = false, defaultValue = "1") Long current,
                                  @RequestParam(value = "rows", required = false, defaultValue = "10") Long rows) {
        Page<Notify> page = new Page<>(current, rows);
        return notifyService.queryNotifys(page);
    }

}
